package Demo1;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class DemoBlotAdd implements IRichBolt{
	int iAddStep;
	int iSumResult;
	OutputCollector _collector;

	public DemoBlotAdd(int i){
		iAddStep = i;
		iSumResult = 0;
	}
	
	@Override
	public void cleanup() {
	}

	@Override
	public void execute(Tuple arg0) {
		String sCommand = arg0.getString(0);
		if (sCommand.equals("a")){
			iSumResult += iAddStep;
		}else{
			iSumResult -= iAddStep;
		}
		String sPrint = "Level--" + iAddStep + ", sCommand = " + sCommand + "\t" + iSumResult + "\n";
		_collector.emit(new Values(sPrint));
	}

	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
		_collector = arg2;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		arg0.declare(new Fields("print"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
}
